#ifndef STATUS_SERVER_H
#define STATUS_SERVER_H

#include <muduo/base/Atomic.h>
#include <muduo/base/Logging.h>
#include <muduo/base/Thread.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/InetAddress.h>

#include <boost/bind.hpp>

#include <muduo/base/Types.h>
#include <map>

#include <sdfs/protobuf/Codec.h>
#include <sdfs/protobuf/ProtobufDispatcher.h>
#include <sdfs/protobuf/Message.h>
#include <sdfs/config/TrackerConfig.h>
#include <sdfs/tracker/StorageServerProxy.h>
#include <sdfs/redis/SyncRedisConnector.h>
#include <sdfs/tracker/FileInfoCodec.h>


using namespace boost;

using namespace muduo;
using namespace muduo::net;

using namespace sdfs::protobuf;
using namespace sdfs::redis;

namespace sdfs
{

namespace tracker
{

class StatusServer
{
public:

	StatusServer(EventLoop* loop, const InetAddress& addr)
		:server_(loop, addr, "StatusServer"),
		  dispatcher_(
		 	boost::bind(&StatusServer::discardMessage, this, _1, _2, _3)
		 ),
		 codec_(
		 	boost::bind(&ProtobufDispatcher::onProtobufMessage, &dispatcher_, _1, _2, _3)
		 ),
		 redisConn_(sdfs::TrackerConfig::redisIp().c_str(), sdfs::TrackerConfig::redisPort()),
		 startTime_(Timestamp::now()),
		 defaultMessageCallback_(
		 	boost::bind(&StatusServer::discardMessage, this, _1, _2, _3))
	{
		proxy_ = StorageServerProxy::getInstance(loop, &codec_ );

		server_.setConnectionCallback(
			boost::bind(&StatusServer::onConnection, this, _1));
		server_.setMessageCallback(
			boost::bind(&ProtobufCodec::onMessage, &codec_, _1, _2, _3));

		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::HeartBeat>(
			boost::bind(&StorageServerProxy::updateStorageServer, proxy_, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileUpdateMessage>(
			boost::bind(&StatusServer::onFileUpdate, this, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileFindMessage>(
			boost::bind(&StatusServer::onFindFile, this, _1, _2, _3));
	}

	~StatusServer()
	{
		if(proxy_ != NULL)
		{
			delete proxy_;
			proxy_ = NULL;
		}
	}

	void onFileUpdate(const muduo::net::TcpConnectionPtr& conn,
		const shared_ptr<FileUpdateMessage> message, muduo::Timestamp timestamp)
	{
		LOG_DEBUG << "fileupdate - sync:" << message->sync() << "from: " <<
		conn->peerAddress().toIpPort();
		if(message->sync())
		{
			muduo::string hashcode = message->hashcode().c_str();
			int groupid = message->groupid();
			StorageServerProxy::getInstance()->addSyncTask(
				groupid, conn, hashcode);
		}
		char cmd[128];
		char value[64];
		snprintf(value, 64, "%s-%d-%ld-%ld", 
			message->storagename().c_str(), message->piece().blockid(),
			message->piece().offset(), message->piece().length());
		snprintf(cmd, 128, "LPUSH %s %s", message->hashcode().c_str(), value);
		redisConn_.command(cmd);
	}

	void onFindFile(const muduo::net::TcpConnectionPtr& conn,
		const shared_ptr<FileFindMessage> message, muduo::Timestamp timestamp)
	{
		char cmd[64];
		snprintf(cmd, 64, "LRANGE %s 0 -1", message->hashcode().c_str());
		RedisReplyPtr reply = redisConn_.command(cmd);
		ServerStatusPtr status;
		FileFindResultMessage result;
		result.set_id(curId_.getAndAdd(1));
		result.set_hashcode(message->hashcode());
		for (int i = 0; i < reply->size(); ++i)
		{
			muduo::string value = reply->getStrAt(i);
			FileInfoCodec codec(value);
			ServerStatusPtr status2 = StorageServerProxy::getInstance()->findStatus(codec.getStorageName());
			if(status2 == NULL)
			{
				continue;
			}
			if(status == NULL || status2 < status)
			{
				status = status2;

				result.set_result(sdfs::protobuf::FileFindResultMessage::FOUND);
				sdfs::protobuf::Worker* worker = result.mutable_worker();
				worker->set_ip(status->getLocalIp().c_str());
				worker->set_port(status->getPort());

				sdfs::protobuf::Filepiece* piece = result.mutable_piece();
				piece->set_blockid(codec.getBlockId());
				piece->set_offset(codec.getOffset());
				piece->set_length(codec.getLength());

				LOG_DEBUG << "find result: "<< status->getLocalIp() << ":" <<status->getPort() << ","
					<<codec.getBlockId()<<","<<codec.getOffset()<<","<<codec.getLength();
			}
		}
		if (status == NULL)
		{
			result.set_result(sdfs::protobuf::FileFindResultMessage::NOTEXIST);
		}
		codec_.Send(conn, result);
	}

	void onConnection(const TcpConnectionPtr& conn)
	{
		LOG_DEBUG << conn->peerAddress().toIpPort() <<
			" -> " << conn->localAddress().toIpPort() <<
			" is " << (conn->connected() ? "UP" : "DOWN");
		if(!conn->connected())
		{
			
			LOG_WARN << "storage: " <<
			 	conn->peerAddress().toIpPort() <<
				" had lost connection!";
			proxy_->removeStorageServer(conn);
		}
	}

	void start()
	{
		LOG_DEBUG << "starting config server";
		server_.start();
		LOG_DEBUG << "connecting redis";
		redisConn_.connect();
	}

	void setDefaultMessageCallback(ProtobufMessageCallback& cb)
	{
		this->defaultMessageCallback_ = cb;
	}

	void discardMessage(const TcpConnectionPtr& conn, 
		const MessagePtr& message, Timestamp timestamp)
	{
		LOG_DEBUG << "MESSAGE NOT DEFINED DISCARD";
	}


private:
	TcpServer server_;
	ProtobufDispatcher dispatcher_;
	ProtobufCodec codec_;
	SyncRedisConnector redisConn_;
	static StorageServerProxy* proxy_;
	Timestamp startTime_;
	ProtobufMessageCallback defaultMessageCallback_;
	int threadNum_;
	AtomicInt32 curId_;
	
};

StorageServerProxy* StatusServer::proxy_ = NULL;

}//tracker

}//sdfs

#endif
